package studio.raptor.extractor.segment.task;

import java.sql.SQLException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.databus.client.DatabusConfig;
import studio.raptor.databus.client.DatabusProducer;
import studio.raptor.databus.client.DatabusProducerBuilder;
import studio.raptor.databus.command.Event;
import studio.raptor.databus.command.EventBasicData;
import studio.raptor.databus.command.EventBusinessData;
import studio.raptor.databus.command.EventBusinessDataProvideType;
import studio.raptor.databus.command.EventBusinessDataType;
import studio.raptor.databus.command.EventMetaData;
import studio.raptor.databus.command.EventMetaDataBusinessType;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.BuildJsonResult;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.service.EventMetaService;

/**
 * Created by liuyong8 on 2017/7/28.
 */
@Component
@Scope("prototype")
public class EventDeliveringTask implements Runnable {

  private static final Logger log = LoggerFactory.getLogger(EventDeliveringTask.class);
  private BlockingQueue<BuildJsonResult> buildJsonQueue;
  private ServerConfig serverConfig;
  private int count;
  private DatabusProducer producer;
  @Autowired
  private EventMetaService eventMetaService;

  @Override
  public void run() {

    try {
      producer = createProducer();
    } catch (Exception e) {
      e.printStackTrace();
    }

    while (!Thread.currentThread().isInterrupted()) {
      BuildJsonResult buildJsonResult = null;
      try {
        buildJsonResult = buildJsonQueue.poll(10, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }

      if (buildJsonResult == null) {
        printPerfTip();
        continue;
      }
      EventMeta eventMeta = buildJsonResult.getEventMeta();
      try {
        //发送报文
        EventBasicData eventBasicData = cover2EventBasicData(eventMeta);
        EventBusinessData eventBusinessData = cover2EventBusinessData(buildJsonResult);
        EventMetaData eventMetaData = cover2EventMetaData(eventMeta);
        Event event = new Event(eventBasicData, eventBusinessData, eventMetaData);
        producer.send(event);
        log.debug("send event {}", event.toString());

        //更新event 为 C
        eventMetaService.modifyEventMetaStatus2C(eventMeta);

      } catch (Exception e) {
        try {
          eventMetaService.modifyEventMetaStatus2F(eventMeta);
          e.printStackTrace();
        } catch (SQLException e1) {
          e1.printStackTrace();
        }
      }


    }

  }

  private DatabusProducer createProducer() throws Exception {
    DatabusConfig config = new DatabusConfig(serverConfig.getBizName());
    DatabusProducerBuilder producerBuilder = new DatabusProducerBuilder(config);
    DatabusProducer producer = producerBuilder.create();
    return producer;
  }

  private EventBasicData cover2EventBasicData(EventMeta eventMeta) {
    EventBasicData eventBasicData = new EventBasicData();
    eventBasicData.setName(serverConfig.getEventName() + "_data");
    eventBasicData.setSource(serverConfig.getBizName());
    eventBasicData.setKindId(serverConfig.getDatabusKindId());
    eventBasicData.setDeliverTime(eventMeta.getDeliverTime());
    return eventBasicData;
  }

  private EventBusinessData cover2EventBusinessData(BuildJsonResult buildJsonResult) {
    EventBusinessData eventBusinessData = new EventBusinessData();
    eventBusinessData.setContent(buildJsonResult.getJson());
    eventBusinessData.setDataService("1");
    eventBusinessData.setDataType(EventBusinessDataType.DATATYPE_JSON);
    eventBusinessData.setProvideType(EventBusinessDataProvideType.PROVIDE_TYPE_CARRY);
    return eventBusinessData;
  }

  private EventMetaData cover2EventMetaData(EventMeta eventMeta) {
    EventMetaData eventMetaData = new EventMetaData();
    //private String businessNo;
    eventMetaData.setBusinessNo(eventMeta.getObjectKey());
    //private EventMetaDataBusinessType businessType;
    eventMetaData.setBusinessType(
        eventMeta.getEventType() == 0 ? EventMetaDataBusinessType.BUSINESS_TYPE_NRM
            : EventMetaDataBusinessType.BUSINESS_TYPE_SAR);
    //private Long regionId;
    eventMetaData.setRegionId(eventMeta.getRegionId());
    //private String prodIds;
    eventMetaData.setProdIds(eventMeta.getProdIds());
    //private String offerIds;
    eventMetaData.setOfferIds(eventMeta.getOfferIds());
    //private String offerSpecAttrs;
    eventMetaData.setOfferSpecAttrs(eventMeta.getOfferSpecAttr());
    //private String serviceOfferIds;
    eventMetaData.setServiceOfferIds(eventMeta.getServiceOfferIds());
    //private String serviceSpecAttrs;
    eventMetaData.setServiceSpecAttrs(eventMeta.getServiceSpecAttr());
    //private Long[] offerProdId;
    eventMetaData.setOfferProdId(null);
    //private String orderTypeIds;
    eventMetaData.setOrderTypeIds(eventMeta.getOrderTypeId());
    //private Long[] custId;
    eventMetaData.setCustId(eventMeta.getCustId());
    //private Long[] acctId;
    eventMetaData.setAcctId(eventMeta.getAcctId());
    //private Long[] prodInstId;
    eventMetaData.setProdInstId(eventMeta.getProdInstId());
    //private Long[] offerInstId;
    eventMetaData.setOfferInstId(eventMeta.getOfferInstId());
    return eventMetaData;
  }

  private void printPerfTip() {
    if ((++count) % 1000 == 0) {
      log.warn("builded events have been delivered, please speed up the building");
      count = 0;
    }
  }

  public void setBuildJsonQueue(BlockingQueue<BuildJsonResult> buildJsonQueue) {
    this.buildJsonQueue = buildJsonQueue;
  }

  public void setServerConfig(ServerConfig serverConfig) {
    this.serverConfig = serverConfig;
  }
}
